package view;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.ObjectOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Iterator;
import java.util.List;
import java.util.Observable;
import java.util.Observer;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.mortbay.util.ajax.Continuation;

import serverMessage.ServerMessage;
import controller.ExecutorsManager;
import controller.ExecutorsManager.AddObserverUsersStatusExecutor;

/**
 * class SendMessagesToUserSocket - singleton,observer - answer to users for sending them updates for their dropbox
 */
public class SendMessagesToUserSocket implements Observer, Runnable {
	
	private int maxOfExecutors;
	private ExecutorService clientRequestPool; //thread pool for handle a client after accept him
	boolean finish=false;
	ServerSocket serverSocket=null;
	
	//the singleton
	private static SendMessagesToUserSocket sendMessagesToUserSocket = new SendMessagesToUserSocket();
	
	//sign as observer for changes in the users messages queues
	private AddObserverUsersStatusExecutor addObserverUsersStatus;
	//the users messages queues - that get from the observer
	private ConcurrentHashMap<String, ConcurrentLinkedQueue<ServerMessage>> userStatusHash;
	
	//save the continuation of each user
	private ConcurrentHashMap<String, Continuation> userContinuationHash = new ConcurrentHashMap<String, Continuation>();
	
	//private ctor
	private SendMessagesToUserSocket()
	{
		maxOfExecutors=10;
		clientRequestPool= Executors.newFixedThreadPool(maxOfExecutors);
		addObserverUsersStatus = ExecutorsManager.getInstance().getAddObserverUsersStatusExecutor();
	}
	
	/**
	 * @return an instance of SendMessagesToUserSocket class (singleton)
	 */
	public static SendMessagesToUserSocket getInstance()
	{
		return sendMessagesToUserSocket;	
	}
	
	/**
	 * add user continuation and suspend it
	 * @param userID - a user name
	 * @param c - the user Continuation
	 */
	public void addContinuation(String userID, Continuation c) {
		
		ConcurrentLinkedQueue<ServerMessage> messageQueue = userStatusHash.get(userID);
			
		//if the user have no messages
		if ((messageQueue==null) || (messageQueue.isEmpty()))
		{
			System.out.println("add continuation userID " + userID + " c " + c);
			userContinuationHash.put(userID, c);
			c.suspend(0);
		}
		
	}

	@Override
	/**
	 * when the observable update its observer - mean new message arrived to a user
	 * and the server need to tell him about it, so the server resume the user continuation
	 * as a sign for the user that he has messages 
	 */
	public void update(Observable o, Object arg) {
		// TODO Auto-generated method stub
		
		System.out.println("update method");
		
		//list of users that get new message
		List<String> list = (List<String>) arg;
		
		Iterator<String> iterator = list.iterator();
		String userID;
		
		//for each user - resume its continuation
		while (iterator.hasNext())
		{
			userID = iterator.next();
			
			//get the user continuation
			Continuation c = userContinuationHash.get(userID);
			
			System.out.println("find continuation userID " + userID + " c " + c);
			if (c!=null)
			{
				try
				{
					System.out.println("doing resume");
					c.resume();
				}
				catch (IllegalStateException e)
				{
					System.out.println("resume error");
					e.printStackTrace();
				}
			}
		}
	}
	
	
	@Override
	/**
	 * answer to users for sending them updates for their dropbox
	 */
	public void run() {
		
		ObjectOutputStream out = null;
		BufferedReader in = null;
	    	    
	    try { 
	    	serverSocket = new ServerSocket(5555);
	    	System.out.println("Server started");
	    }
	    catch (IOException e) { 
	    	System.err.println("Cannot listen on port 5555");
	    	return;
	    	//System.exit(-1);
	    }
	    
	    Socket clientSocket=null;
	    while(!finish){
		    try { 
		    	System.out.println("accept client socket 5555");
		    	//answer a client
		    	clientSocket=serverSocket.accept();
		    	System.out.println("accepted client socket 5555");
		    	out = new ObjectOutputStream(clientSocket.getOutputStream());
		    	//InputStream is = clientSocket.getInputStream();
		    	in=  new BufferedReader(new InputStreamReader(clientSocket.getInputStream())); 
		    	
		    	//for free this thread to listen to another requests
		    	clientRequestPool.execute(new MyProtocolSendMessage(in,out,clientSocket,userStatusHash));
		    }
		    catch (IOException e) {
		    	System.err.println("Accept failed");
		    	e.printStackTrace();
		    	
		    }
	    }
	    
		try {
			//close the server socket
			if ((serverSocket!=null)&&(!serverSocket.isClosed()))
				serverSocket.close();
			
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}	
		
	}

	
	/**
	 * sign himself as observer for updates in the users messages queues
	 */
	public void signAsObserver()
	{
		userStatusHash = addObserverUsersStatus.addObserverUsersStatus(this);
	}

	
	/**
	 * stop to send messages to users
	 */
	public void stop() {
		this.finish = true;
		clientRequestPool.shutdown();
		
		try {
			if ((serverSocket!=null)&&(!serverSocket.isClosed()))
				serverSocket.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	

}
